package com.wz.jiangsu.main.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


public class App {
    public static void main(String[] args) {

        long l = System.currentTimeMillis();
        System.out.println(l);
//        String subTopic = "testtopic/#";
//        String pubTopic = "testtopic/1";
        String subTopic = "sl/#";
        String pubTopic = "sl/2";
//        String content = "wz_table4,host=pop-os,size=10 value=100i 1653579153147395661";
        int qos = 1;
        String broker = "tcp://10.25.10.149:1883";
        String clientId = "Telegraf-Consumer-B9WAO1";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient client = new MqttClient(broker, clientId, persistence);

            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName("admin");
            connOpts.setPassword("Netvine@123".toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);

            // 设置回调
            client.setCallback(new OnMessageCallback());

            // 建立连接
            System.out.println("Connecting to broker: " + broker);
            client.connect(connOpts);

            System.out.println("Connected");
//            System.out.println("Publishing message: " + content);

            // 订阅
            client.subscribe(subTopic);


            for (int i = 0; i < 10000; i++) {

                long currentTimeInNanos = System.currentTimeMillis() * 1000000;

                // 使用 String.format 将时间戳插入到字符串中
                String content = String.format("wz_table_19,host=pop-os1,size=56 value=45i %d", currentTimeInNanos);

                // 消息发布所需参数
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                // 获取当前的Unix时间戳，单位为纳秒

                client.publish(pubTopic, message);
            }

            System.out.println("Message published");

            client.disconnect();
            System.out.println("Disconnected");
            client.close();
            System.exit(0);
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

class OnMessageCallback implements MqttCallback {
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连");
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        System.out.println("接收消息主题:" + topic);
        System.out.println("接收消息Qos:" + message.getQos());
        System.out.println("接收消息内容:" + new String(message.getPayload()));
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }
}
